package graph

import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CommitFriends {
  def main(args: Array[String]): Unit = {
    //saprkContext
    val conf = new SparkConf().setMaster("local[*]").setAppName(s"${this.getClass.getName}")
    val sc = new SparkContext(conf)
    //读取数据
    val lines = sc.textFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\图计算案例2.txt")
    //创建点集合
    val pointRDD: RDD[(Long, String)] = lines.flatMap(line => {
      val pointArr: Array[String] = line.split("\t")
      pointArr.map(tp => (tp.hashCode.toLong, tp))
    }).distinct()
    //创建边集合
    val edgeRDD: RDD[Edge[Int]] = lines.flatMap(line => {
      val pointArr = line.split("\t") //王五 赵六  //李四
      //pointArr.tail.zip(pointArr).map(tp => Edge(tp._1.hashCode.toLong, tp._2.hashCode.toLong, 0))
      println(pointArr.tail.size)
      pointArr.map(tp=>Edge(pointArr.head.hashCode.toLong,tp.hashCode.toLong,0))
    })
    edgeRDD


    //创建图对象
    val graph = Graph(pointRDD,edgeRDD)
    val ver = graph.connectedComponents().vertices
    ver.join(pointRDD).map{
      case (id,(mid,name)) => (mid,name)
    }.reduceByKey((a,b)=>a.concat(",").concat(b)).foreach(println)
    //释放资源
    sc.stop()
  }
}
